#ifndef A8_SOCKET_H
#define A8_SOCKET_H

#include <string.h>
#include <string>
#include <list>
#include <errno.h>
#include <sys/socket.h>
#include <a8/delegate.hpp>
#include <a8/thread.hpp>
#include <a8/lock.hpp>

#ifdef _WIN32   
typedef int socklen_t;  
typedef int ssize_t;   
#endif   
#ifdef __linux__   
typedef int SOCKET;   
#define SOCKET_ERROR (-1)   
#define INVALID_SOCKET (-1)
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <strings.h>
#endif   

namespace a8
{
  struct SendQueueNode
  {
    char* mBuff;
    int mBuffLen;
    int mStartPos;
    SendQueueNode *mNext;
  };

  class TcpClient
  {
  public:
    a8::Delegate<void (TcpClient*, int)> OnError;
    a8::Delegate<void (TcpClient*)> OnConnect;
    a8::Delegate<void (TcpClient*)> OnDisConnect;
    a8::Delegate<void (TcpClient*, char*, int)> OnSocketRead;

    TcpClient():mSocket(INVALID_SOCKET), mConnected(false), mRemoteAddress(""), mRemotePort(0),
		mWorkerThread(NULL), mFirstNode(NULL), mLastNode(NULL)
    {
    }
    virtual ~TcpClient()
    {
      if(mWorkerThread)
	ActiveStop();
      mSocket = INVALID_SOCKET;
    }
    SOCKET getSocket() { return mSocket; };
    bool isActive() { return mSocket != INVALID_SOCKET; }    
    bool Connected() { return mConnected; }

    void setActive(bool active) 
    { 
      if(active){
	if(isActive())
	  return;
	if(!mWorkerThread){
	  mWorkerThread = new a8::Runnable();
	  mWorkerThread->OnExecute += std::make_pair(this, &TcpClient::WorkerThreadProc)
;
	  mWorkerThread->Start();
	}
      }else{
	if(mWorkerThread)
	  ActiveStop();
	mSocket = INVALID_SOCKET;
      }
    }

  private:
    SOCKET mSocket;
    bool mConnected;
    std::string mRemoteAddress;
    int mRemotePort;
    a8::Runnable* mWorkerThread;
    a8::Mutex mSendBufferMutex;
    SendQueueNode *mFirstNode;
    SendQueueNode *mLastNode;

    bool ActiveStart()
    {
      mSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
      if(INVALID_SOCKET == mSocket){
	OnError(this, errno);
	return false;
      }
      sockaddr_in sa;
      bzero(&sa, sizeof(sa));
      sa.sin_family = AF_INET;
      sa.sin_addr.s_addr = inet_addr(mRemoteAddress.c_str());
      sa.sin_port = htons(mRemotePort);
      if(connect(mSocket, (sockaddr*)&sa, sizeof(sa)) < 0){
	OnError(this, errno);
	return false;
      }
      fcntl(mSocket, F_SETFL, O_NONBLOCK);
      OnConnect(this);
      return true;
    }
    void ActiveStop()
    {
      if(mWorkerThread){
	delete mWorkerThread;
	mWorkerThread = NULL;
      }
      mSocket = INVALID_SOCKET;
    }
    void WorkerThreadProc(a8::Runnable* thread)
    {
      if(!ActiveStart())
	return;
      fd_set rfds, wfds;
      while(!thread->IsShutdown()){
	FD_ZERO(&rfds);
	FD_ZERO(&wfds);
	FD_SET(mSocket, &rfds);
	FD_SET(mSocket, &wfds);
	switch(select(mSocket+1, &rfds, &wfds, NULL, 0)){
	case -1:  //error
	  break;
	case 0:   //timeout
	  break;
	default:
	  if(FD_ISSET(mSocket, &rfds))
	    ReadyRecv();
	  if(FD_ISSET(mSocket, &wfds))
	    ReadySend();
	}
      }
    }
    void ReadyRecv()
    {
      char inBuf[1024];
      while(1){
	int ret = recv(mSocket, inBuf, 1024, 0);
	if(ret < 0){
	  if(errno != EAGAIN){
	    OnError(this, errno);
	    mWorkerThread->Stop();
	  }
	  break;
	}else if(ret == 0){
	  OnDisConnect(this);
	  mWorkerThread->Stop();
	  break;
	}else{
	  OnSocketRead(this, inBuf, ret);
	  if(ret < 1024)
	    break;
	}
      }
    }
    void ReadySend()
    {
      mSendBufferMutex.Lock();
      __try{
	while(mFirstNode){
	  char *szData = mFirstNode->mBuff + mFirstNode->mStartPos;
	  int iDataLen = mFirstNode->mBuffLen - mFirstNode->mStartPos;
	  int iRet = send(mSocket, szData, iDataLen, 0);
	  if(iRet >= 0){
	    if(iRet == iDataLen){
	      mFirstNode = mFirstNode->mNext;
	      if(!mFirstNode)
		mLastNode = NULL;
	    }else
	      mFirstNode->mStartPos += iDataLen;
	  }else{
	    OnError(this, errno);
	    mWorkerThread->Stop();
	    return;
	  }
	}
      }catch(...){
      
      }/*finally{
	mSendBufferMutex.UnLock();
	}*/
    }
  };

  class ClientSocket
  {
  public:
    a8::Delegate<void (ClientSocket*, int)> OnError;
    a8::Delegate<void (ClientSocket*)> OnConnect;
    a8::Delegate<void (ClientSocket*)> OnDisConnect;
    a8::Delegate<void (ClientSocket*, char*, int)> OnSocketRead;

    ClientSocket():mSocket(INVALID_SOCKET), mRemoteAddress(""), mRemotePort(0), mFirstNode(NULL), mLastNode(NULL), mSocketID(0)
    {
    }
    virtual ~ClientSocket()
    {
    }
    void ReadyRecv()
    {
      char inBuf[1024];
      while(1){
	int ret = recv(mSocket, inBuf, 1024, 0);
	if(ret < 0){
	  if(errno != EAGAIN){
	    OnError(this, errno);
	  }
	  break;
	}else if(ret == 0){
	  OnDisConnect(this);
	  break;
	}else{
	  OnSocketRead(this, inBuf, ret);
	  if(ret < 1024)
	    break;
	}
      }
    }
    void ReadySend()
    {
      mSendBufferMutex.Lock();
      /*try*/{
	while(mFirstNode){
	  char *szData = mFirstNode->mBuff + mFirstNode->mStartPos;
	  int iDataLen = mFirstNode->mBuffLen - mFirstNode->mStartPos;
	  int iRet = send(mSocket, szData, iDataLen, 0);
	  if(iRet >= 0){
	    if(iRet == iDataLen){
	      mFirstNode = mFirstNode->mNext;
	      if(!mFirstNode)
		mLastNode = NULL;
	    }else
	      mFirstNode->mStartPos += iDataLen;
	  }else{
	    OnError(this, errno);
	    return;
	  }
	}
      }/*finally*/{
	mSendBufferMutex.UnLock();
      }
    }
  private:
    unsigned short mSocketID;
    SOCKET mSocket;  
    std::string mRemoteAddress;
    int mRemotePort;
    a8::Mutex mSendBufferMutex;
    SendQueueNode *mFirstNode;
    SendQueueNode *mLastNode;
  };

  class TcpListener
  {
  public:
    a8::Delegate<void (TcpListener*, int errorId)> OnError;
    
    TcpListener(int maxClientCnt):mListenSocket(INVALID_SOCKET), mAddress(""), mPort(0),
				  mAcceptThread(NULL), mWorkerThread(NULL),  
				  mMaxClientCnt(maxClientCnt)
    {
      #ifdef __linux__
      mEpollFD = epoll_create(mMaxClientCnt);
      #endif
    }
    virtual ~TcpListener()
    {
      #ifdef __linux__
      epoll_destroy(mEpollFD);
      #endif
    }
 
    virtual ClientSocket* CreateClientSocket() { return new ClientSocket();}
    SOCKET getSocket() { return mListenSocket; }
    bool isActive() {return mListenSocket != INVALID_SOCKET;}
    std::string getAddress() {return mAddress;}
    void setAddress(const std::string& ipAddress) { mAddress = ipAddress;}
    int getPort() {return mPort;}
    void setPort(int port) { mPort = port;}    
    void setActive(bool active) 
    { 
      if(active){
	if(isActive())
	  return;
	if(!mWorkerThread){
	  mWorkerThread = new Runnable();
	  mWorkerThread->OnExecute += std::make_pair(this, &TcpListener::WorkerThreadProc);
	  mWorkerThread->Start();
	}
      }else{
	if(mWorkerThread)
	  ActiveStop();
	mListenSocket = INVALID_SOCKET;
      }
    }
  private:
    SOCKET mListenSocket;
    std::string mAddress;
    int mPort;   
    a8::Runnable* mAcceptThread;
    a8::Runnable* mWorkerThread;
    a8::Mutex mClientsMutex;
    std::list<ClientSocket*> mClientList;
    int mMaxClientCnt;
    #ifdef __linux__
    int mEpollFD;
    #endif

    void AcceptThreadProc(a8::Runnable* thread)
    {
      if(listen(mListenSocket, 65535) < 0) {
	OnError(this, errno);
	return;
      }

      while(!thread->IsShutdown()){
	  if(INVALID_SOCKET == mListenSocket)
	    break;
	  socklen_t addrlen = sizeof(sockaddr_in);
	  sockaddr_in addr;
	  memset(&addr, 0, sizeof(addr));
	  SOCKET sock = accept(mListenSocket, (sockaddr*)&addr, &addrlen);
	  if(thread->IsShutdown())
	    break;
	  if(INVALID_SOCKET != sock){
	    SocketAccept(sock, inet_ntoa(addr.sin_addr), addr.sin_port);
	  }else{
	    OnError(this, errno);
	  }
      }
    }

    void WorkerThreadProc(a8::Runnable* thread)
    {
      if(!ActiveStart())
	return;
      #ifdef _WIN32
      #endif
      #ifdef __linux__
      epoll_event *events = new epoll_event[mMaxClientCnt];
      while(!thread->IsShutdown()){
	int nfds = epoll_wait(mEpollFD, events, mMaxClientCnt, -1);
	for(int i = 0; i < nfds; i++){	  
	  if(events[i].events & EPOLLIN){
	    ClientSocket *client = (ClientSocket*)events[i].data.ptr;
	    client->ReadyRecv();
	  }else if(events[i].events & EPOLLOUT){
	    ClientSocket *client = (ClientSocket*)events[i].data.ptr;
	    client->ReadySend();
	  }
	}
      }
      delete [] event;
      #endif
      #ifdef _freebsd_
      #endif
    }

    bool ActiveStart()
    {
      mListenSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
      if(INVALID_SOCKET == mListenSocket){
	OnError(this, errno);
	return false;
      }
      sockaddr_in sa;
      bzero(&sa, sizeof(sa));
      sa.sin_family = AF_INET;
      sa.sin_addr.s_addr = inet_addr(mAddress.c_str());
      sa.sin_port = htons(mPort);
      if(bind(mListenSocket, (sockaddr*)&sa, sizeof(sa)) < 0){
	OnError(this, errno);
	return false;
      }
      mAcceptThread = new a8::Runnable();
      mAcceptThread->OnExecute += std::make_pair(this, &TcpListener::AcceptThreadProc);
      mAcceptThread->Start();
      return true;
    }
    void ActiveStop()
    {
    }
    void SocketAccept(SOCKET sock, char* ipAddr, int port)
    {
      mClientsMutex.Lock();
      
 
      mClientsMutext.UnLock();
    }
  };
}

#endif
